package com.ac.scum.log.task.face.log;

import com.ac.common.core.exception.BaseException;
import com.ac.scum.log.bean.Server;
import com.ac.scum.log.bean.ServerFTP;
import com.ac.scum.log.bean.ServerFTPFIle;
import com.ac.scum.log.task.mapper.server.ServerFTPFileMapper;
import com.ac.scum.log.task.mapper.server.ServerFTPMapper;
import com.ac.scum.log.task.service.ServerService;
import com.ac.scum.log.task.utils.ftp.FTPFile;
import com.ac.scum.log.task.utils.ftp.FTPUtil;
import com.ac.scum.log.task.utils.queue.FTPQueueItem;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.google.common.cache.CacheBuilder;
import com.google.common.cache.CacheLoader;
import com.google.common.cache.LoadingCache;
import com.google.common.collect.Lists;
import com.jcraft.jsch.JSchException;
import com.jcraft.jsch.SftpException;
import io.micrometer.common.util.StringUtils;
import jakarta.annotation.PostConstruct;
import jakarta.annotation.PreDestroy;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.time.LocalDateTime;
import java.util.*;
import java.util.concurrent.*;

@Slf4j
@Component
@AllArgsConstructor
public class LogUpdateProvider {
    private final ServerService serverService;
    private final ServerFTPMapper serverFTPMapper;
    private final ServerFTPFileMapper serverFtpFileMapper;
    private final StringRedisTemplate stringRedisTemplate;
    private final RedisTemplate<String, Object> queueRedisTemplate;
    public static final String DATA_CHECK_CACHE_KEY = "QUEUE_DATA_CHECK_CACHE";
    public static final String REDIS_QUEUE_KEY = "LOG_UPDATE_QUEUE";
    private static ThreadPoolExecutor threadPool;
    private final RedissonClient redissonClient;
    private final LoadingCache<Long, List<Server>> serverCache = CacheBuilder.newBuilder()
            .expireAfterWrite(10, TimeUnit.SECONDS)
            .build(new CacheLoader<>() {
                @SuppressWarnings("NullableProblems")
                @Override
                public List<Server> load(Long packageId) {
                    return serverService.getServersWithSpecifiedPackageIds(packageId);
                }
            });

    static {
        threadPool = new ThreadPoolExecutor(
                20,
                300,
                60,
                TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(300),
                new ThreadPoolExecutor.CallerRunsPolicy());

    }

    @PostConstruct
    public void serverLogsUpdate() {
        Timer timer = new Timer();
        TimerTask task = new TimerTask() {
            @Override
            public void run() {
                if (threadPool.isShutdown() || threadPool.isTerminated() || threadPool.isTerminating()) {
                    log.warn("线程池当前处于非运行状态，跳过本次任务执行");
                    return;
                }
                List<Server> serversWithSpecifiedPackageIds = Lists.newArrayList();
                try {
                    serversWithSpecifiedPackageIds = serverCache.get(1L);
                } catch (ExecutionException e) {
                    log.error("从缓存获取有套餐的服务器失败: {}", e.getMessage());
                }
                for (Server server : serversWithSpecifiedPackageIds) {
                    RLock lock = redissonClient.getLock("log-upd-task::" + server.getId());
                    if (lock.isLocked()) {
                        log.info("当前服务器：{} 正在更新日志，跳过本次任务执行", server.getName());
                        continue;
                    }
                    threadPool.execute(() -> updateLogs(server));
                }
            }
        };
        // 每隔20秒执行一次
        timer.schedule(task, 0, 20000);
    }

    @Scheduled(cron = "0 0 8/14 * * *")
    public void scheduleTask() {
        tobeShutdown();
        threadPool = new ThreadPoolExecutor(
                20,
                300,
                60,
                TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(300),
                new ThreadPoolExecutor.CallerRunsPolicy());
        log.info("线程池已重新初始化");
    }

    @PreDestroy
    @SuppressWarnings("DuplicatedCode")
    public void tobeShutdown() {
        try {
            log.info("开始关闭提供者线程池");
            threadPool.shutdown();

            if (!threadPool.awaitTermination(3, TimeUnit.MINUTES)) {
                log.error("提供者线程池未能在30秒内完全关闭，执行强制关停");
                for (Runnable runnable : threadPool.shutdownNow()) {
                    log.error("提供者线程池未处理的任务：{}", runnable);
                }
            }
        } catch (InterruptedException e) {
            log.error("提供者线程池关闭过程中被打断", e);
        }
        log.info("提供者线程池已关闭");
    }

    @SuppressWarnings("DuplicatedCode")
    public void updateLogs(Server server) {
        RLock lock = redissonClient.getLock("log-upd-task::" + server.getId());
        if (lock.isLocked()) {
            return;
        }
        FTPUtil ftpUtil = null;
        long startTime = System.currentTimeMillis();
        try {
            lock.lock(10, TimeUnit.MINUTES);
            boolean longStorageFlag = serverFTPMapper.findValidPackage(server.getId(), 2L) != null;

//            log.info("⭐服务器：[{}]日志任务更新任务于{}开始 类型：{}", server.getName(), startTime, (longStorageFlag ? "长久存储" : "临时存储"));

            ServerFTP serverFTP = serverFTPMapper.selectOne(Wrappers.lambdaQuery(ServerFTP.class).eq(ServerFTP::getServerId, server.getId()));
            if (serverFTP == null) {
                log.info("服务器：[{}]日志任务更新任务终止，原因:未获取到FTP/SFTP相关配置", server.getName());
                return;
            }

            ftpUtil = new FTPUtil(serverFTP);
            try {
                ftpUtil.connect();
            } catch (IOException | JSchException e) {
                connectionFail(e, serverFTP, ftpUtil, server);
                return;
            }


            List<FTPFile> ftpFileList = ftpUtil.listFiles(serverFTP.getLogPath());
            if (ftpFileList.isEmpty()) {
                log.warn("服务器：[{}]日志任务更新任务终止，原因:为获取到文件列表", server.getName());
                return;
            }
            List<ServerFTPFIle> serverFtpFileList = serverFtpFileMapper.selectList(Wrappers.lambdaQuery(ServerFTPFIle.class)
                    .eq(ServerFTPFIle::getServerId, server.getId())
                    .in(ServerFTPFIle::getFileName, ftpFileList.stream().map(FTPFile::getName).toList()));
            for (FTPFile ftpFile : ftpFileList) {
                Optional<ServerFTPFIle> dbFileOptional = serverFtpFileList.stream().filter(item -> item.getFileName().equals(ftpFile.getName())).findFirst();
                String cacheName = server.getId() + ftpFile.getName();
                String cacheSize = (String) stringRedisTemplate.opsForHash().get(DATA_CHECK_CACHE_KEY, cacheName);
                //noinspection SpellCheckingInspection
                if (ftpFile.getSize() <= 98 || ftpFile.getName().startsWith("loot") || ftpFile.getName().startsWith("famepoints_") || !ftpFile.getName().endsWith(".log")) {
                    log.debug("服务器：[{}]日志任务更新任务 Rule Skip File {}", server.getName(), ftpFile.getName());
                } else if (cacheSize != null) {
                    log.debug("服务器：[{}]日志任务更新任务 Repeat Skip File {} Size {}", server.getName(), ftpFile.getName(), ftpFile.getSize());
                } else if (dbFileOptional.isEmpty()) {
                    FTPUtil finalFtpUtil = ftpUtil;
                    Future<Boolean> submit = threadPool.submit(() -> commitQueue(server, ftpFile, serverFTP, finalFtpUtil, 0L, null, longStorageFlag));
                    try {
                        if (!submit.get(10, TimeUnit.SECONDS)) {
                            log.error("服务器：[{}]日志任务更新任务执行失败", server.getName());
                            return;
                        }
                    } catch (InterruptedException | ExecutionException e) {
                        log.error("服务器：[{}]日志任务更新任务执行失败", server.getName(), e);
                        return;
                    } catch (TimeoutException e) {
                        log.error("服务器：[{}]日志任务更新任务执行超时", server.getName());
                        submit.cancel(true);
                        return;
                    }
                } else if (!dbFileOptional.get().getFileSize().equals(ftpFile.getSize())) {
                    ServerFTPFIle serverFTPFIle = dbFileOptional.get();
                    FTPUtil finalFtpUtil = ftpUtil;
                    Future<Boolean> submit = threadPool.submit(() -> commitQueue(server, ftpFile, serverFTP, finalFtpUtil, serverFTPFIle.getFileSize(), serverFTPFIle.getId(), longStorageFlag));

                    try {
                        if (!submit.get(10, TimeUnit.SECONDS)) {
                            log.error("服务器：[{}]日志任务更新任务执行失败", server.getName());
                            return;
                        }
                    } catch (InterruptedException | ExecutionException e) {
                        log.error("服务器：[{}]日志任务更新任务执行失败", server.getName(), e);
                        return;
                    } catch (TimeoutException e) {
                        log.error("服务器：[{}]日志任务更新任务执行超时", server.getName());
                        submit.cancel(true);
                        return;
                    }
                } else {
                    log.debug("服务器：[{}]日志任务更新任务 NoChange Skip File {}", server.getName(), ftpFile.getName());
                }
            }
            serverFTP.setStatus(1);
            serverFTP.setLastConnectMsg("SUCCESS");
            serverFTP.setLastConnectTime(LocalDateTime.now());
            serverFTPMapper.updateById(serverFTP);

        } catch (IOException | SftpException e) {
            log.error("获取文件列表失败 :{}, {}", server.getName(), e.getMessage(), e);
        } finally {
            if (ftpUtil != null) {
                try {
                    ftpUtil.disconnect();
                } catch (RuntimeException e) {
                    log.error("ftp/sftp 关闭失败 : {}", e.getMessage());
                }
            }
            if (lock.isLocked()) {
                lock.unlock();
            }
        }
        log.info("⭐服务器：[{}]日志任务更新任务于{}结束，共耗时{}s", server.getName(), startTime, (System.currentTimeMillis() - startTime) / 1000);
    }


    private boolean commitQueue(Server server, FTPFile ftpFile, ServerFTP serverFTP, FTPUtil ftpUtil, Long skipSize, Long fileId, Boolean storageFlag) {
        FTPQueueItem ftpQueueItem = new FTPQueueItem();
        ftpQueueItem.setServer(server);
        ftpQueueItem.setServerFTP(serverFTP);
        ftpQueueItem.setName(ftpFile.getName());
        ftpQueueItem.setFileId(fileId);
        ftpQueueItem.setIsNew(fileId == null);
        ftpQueueItem.setOffset(skipSize);
        ftpQueueItem.setSize(ftpFile.getSize());
        ftpQueueItem.setLongStorageFlag(storageFlag);
        try {
            String path = serverFTP.getLogPath() + "/" + ftpFile.getName();
            ftpUtil.processFileInputStream(path, inputStream -> {
                long start = System.currentTimeMillis();
                ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
                long skip = inputStream.skip(skipSize);
                log.debug("⭐服务器：[{}]日志任务-提交队列 ： 跳过{}字节", server.getName(), skip);
                byte[] buffer = new byte[1024];
                int len;
                while ((len = inputStream.read(buffer, 0, buffer.length)) != -1) {
                    outputStream.write(buffer, 0, len);
                    Arrays.fill(buffer, (byte) 0);
                }
                ftpQueueItem.setContext(outputStream.toString(StandardCharsets.UTF_16LE));
                log.info("⭐服务器：[{}]日志任务-提交队列 : {} 获取字节流成功 Size {}b 耗时 {}ms", server.getName(), ftpFile.getName(), ftpFile.getSize() - skipSize, (System.currentTimeMillis() - start));
            });
        } catch (IOException | SftpException | BaseException e) {
            log.error("服务器：[{}]日志任务-提交队列 : 获取{}文件流失败", server.getName(), ftpFile.getName(), e);
            return false;
        }
        if (StringUtils.isNotBlank(ftpQueueItem.getContext())) {
            String cacheName = server.getId() + ftpFile.getName();
            stringRedisTemplate.opsForHash().put(DATA_CHECK_CACHE_KEY, cacheName, ftpFile.getSize().toString());
            queueRedisTemplate.opsForList().rightPush(REDIS_QUEUE_KEY, ftpQueueItem);
        }
        return true;
    }

    private void connectionFail(Exception e, ServerFTP serverFTP, FTPUtil ftpUtil, Server server) {
        serverFTP.setStatus(2);
        serverFTP.setLastConnectMsg(e.getMessage());
        serverFTPMapper.updateById(serverFTP);
        ftpUtil.disconnect();
        log.info("⭐⭐⭐⭐⭐⭐⭐⭐⭐⭐⭐⭐⭐⭐⭐服务器：[{}]日志任务连接失败,日志已提交⭐⭐⭐⭐⭐⭐⭐⭐⭐⭐⭐⭐⭐⭐⭐", server.getName());

    }
}
